﻿using Apache.IO;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace Reflexive.IO
{

    public class KafkaLogFileManager : IList<long> 
    {
        DurableFileStream _transactionLog;

        public KafkaLogFileManager (string filename):base()
        {
            _transactionLog = new DurableFileStream(filename, false);
        }

        
        public IEnumerator<long> GetEnumerator()
        {
            byte[] buffer = new byte[sizeof(long)];
            _transactionLog.Seek(0, System.IO.SeekOrigin.Begin);
            return new long[Count].Select((offset, i) =>
            {
                buffer = new byte[sizeof(long)];
                _transactionLog.Read(buffer, 0, sizeof(long));
                return BitConverter.ToInt64(buffer, 0);
            }).GetEnumerator();
        }

        System.Collections.IEnumerator System.Collections.IEnumerable.GetEnumerator()
        {
            return GetEnumerator();
        }


        public int IndexOf(long value)
        {
            return this.Select((offset, i) => { return offset == value ? i : -1; }).DefaultIfEmpty(-1).FirstOrDefault();
        }

        public void Insert(int index, long value)
        {
            byte[] buffer = BitConverter.GetBytes(value);
            _transactionLog.Seek(index * sizeof(long), System.IO.SeekOrigin.Begin);
            _transactionLog.Write(buffer,0,buffer.Length);
            _transactionLog.Commit();
        }

        public void RemoveAt(int index)
        {
            byte[] buffer = BitConverter.GetBytes(0);
            _transactionLog.Seek(index * sizeof(long), System.IO.SeekOrigin.Begin);
            _transactionLog.Write(buffer, 0, buffer.Length);
            _transactionLog.Commit();
        }

        public long this[int index]
        {
            get
            {
                var buffer = new byte[sizeof(long)];
                _transactionLog.Seek(index * sizeof(long), System.IO.SeekOrigin.Begin);
                _transactionLog.Read(buffer, 0, sizeof(long));
                return BitConverter.ToInt64(buffer, 0);
            }
            set
            {
                Insert(index, value);
            }
        }

        public void Add(long value)
        {
            byte[] buffer = BitConverter.GetBytes(value);
            _transactionLog.Seek(0, System.IO.SeekOrigin.End);
            _transactionLog.Write(buffer, 0, buffer.Length);
            _transactionLog.Commit();
        
        }

        public void Clear()
        {
           for(int i = 0; i < Count; i++)
           {
               RemoveAt(i);
           }
        }

        public bool Contains(long value)
        {
            return IndexOf(value) > 0;
        }

        public void CopyTo(long[] array, int arrayIndex)
        {
            int count = array.Length - arrayIndex;
            for (var i = arrayIndex; i < array.Length; i++)
            {
                array[i] = this[i];
            }
        }

        public int Count
        {
            get { return (int)_transactionLog.Length / sizeof(long); }
        }

        public bool IsReadOnly
        {
            get { return false; }
        }

        public bool Remove(long value)
        {
            int index = IndexOf(value);
            RemoveAt(index);
            return index > -1;
        }
    }
}
